package org.swift.mashup.engine.concurrency.agenda;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.swift.commons.client.RestCall;
import org.swift.mashup.engine.concurrency.ConcurrencyException;
import org.swift.mashup.engine.concurrency.ConcurrentContext;

public class RestfulJobSet extends JobSet  {
	
	private final static Logger log = Logger.getLogger(RestfulJobSet.class);
	
	public RestfulJobSet(ConcurrentContext context, String url, String processId,
			String workingDirectory, int startindex, int endindex) {
		super( context, url, processId, workingDirectory, startindex, endindex, null);
	}	

	@Override
	public void run() {
		String restURL = null;
		long startTime = System.nanoTime();
		ConcurrentContext ctx = getContext();
		while ((executionAttempts++) < ctx.getConfiguration().getMaxAttempts()){
			try {
				restURL = ctx.getNextComputeNode() + getUrl();
    			setComputeNode(restURL);
    			log.debug("Running with restful jobset on node: " + restURL);
    			RestCall call = new RestCall(restURL, "GET");
    			call.sendRecieve(createParameterMap());
    			long duration = TimeUnit.MILLISECONDS.convert(
    				System.nanoTime()-startTime, TimeUnit.NANOSECONDS);
    			String msg = String.format("Completed job #%s @ %s in %d ms for " +
    					"items starting with startIndex: %d  interval: %d",
    				getProcessId(), restURL, duration, getNewMemoryPosition(), getLimit());
    			log.info(msg);
    			break;
    		} catch (ConcurrencyException ce){
    			throw new RuntimeException ("Error getting next compute node.",ce);
    		} catch (Exception e) {
    			long duration = TimeUnit.MILLISECONDS.convert(
    				System.nanoTime()-startTime, TimeUnit.NANOSECONDS);
    			String msg = String.format("Job #%s failed @ %s in %d ms for " +
    				"items starting with startIndex: %d  interval: %d",
    				getProcessId(), restURL, duration, getNewMemoryPosition(), getLimit());
    			log.error(msg, e);
    			setError(e);
    		}
		}
	}

	protected Map<String, ?> createParameterMap() {
		Map<String, Object> map = new HashMap<String, Object>();
		map.put("processid", this.getProcessId());
		map.put("sessionid", this.getSessionId());
		map.put("workingDirectory", this.getWorkingDirectory());
		map.put("startIndex", Integer.toString(this.getNewMemoryPosition()));
		map.put("endIndex", Integer.toString(this.getLimit()));
		return map;
	}
}
